package com.codrim.cs.data.server.dao.realtime;

import com.codrim.common.utils.json.JsonMapper;
import com.codrim.cs.data.server.dao.RealTimeDataDao;
import com.codrim.cs.data.server.dao.param.RealTimeDataSearchParam;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Repository;

import java.time.Instant;
import java.util.List;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.regex.Pattern;

import javax.annotation.Resource;

/**
 * 实时数据处理 相关数据存储逻辑, 数据类型为ZSET: real_time_gross_conversions: 总转化数量 real_time_scrub_conversions: 被核减的转化数量 real_time_clicks
 * 点击数量 real_time_uclicks 去重点击数量 所有zset中存放的key: advId_affId_offerId_hour   hour为0时区的时间
 *
 * Created by Liang.Zhuge on 27/07/2017.
 */
@Repository
public class RealTimeDataDaoImpl implements RealTimeDataDao {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private static final String REAL_TIME_GROSS_CONVERSIONS = "real_time_gross_conversions";
    private static final String REAL_TIME_SCRUB_CONVERSIONS = "real_time_scrub_conversions";
    private static final String REAL_TIME_CLICKS = "real_time_clicks";
    private static final String REAL_TIME_UCLICKS = "real_time_uclicks";
    static final Pattern KEY_PATTERN = Pattern.compile("(\\d+)_(\\d+)_(\\d+)_(\\d{10})");

    @Resource StringRedisTemplate stringRedisTemplate;

    /**
     * 查询指定广告主时间范围内实时数据
     *
     * @param advId 广告主ID
     * @param begin 查询开始时间
     * @param end   查询结束时间
     */
    @Override
    public List<RealTimeData> advertiserRealTimeData(int advId, Instant begin, Instant end) {
        final RealTimeDataHelper helper = new RealTimeDataHelper();

        stringRedisTemplate.boundZSetOps(REAL_TIME_CLICKS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAdvertiser(advId, begin, end, item, helper::addClicks));
        stringRedisTemplate.boundZSetOps(REAL_TIME_UCLICKS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAdvertiser(advId, begin, end, item, helper::addUClicks));
        stringRedisTemplate.boundZSetOps(REAL_TIME_GROSS_CONVERSIONS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAdvertiser(advId, begin, end, item, helper::addGrossConversions));
        stringRedisTemplate.boundZSetOps(REAL_TIME_SCRUB_CONVERSIONS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAdvertiser(advId, begin, end, item, helper::addScrubConversions));

        return helper.data();
    }

    @Override
    public List<RealTimeData> affiliateRealTimeData(int affId, Instant begin, Instant end) {
        final RealTimeDataHelper helper = new RealTimeDataHelper();

        stringRedisTemplate.boundZSetOps(REAL_TIME_CLICKS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAffiliate(affId, begin, end, item, helper::addClicks));
        stringRedisTemplate.boundZSetOps(REAL_TIME_UCLICKS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAffiliate(affId, begin, end, item, helper::addUClicks));
        stringRedisTemplate.boundZSetOps(REAL_TIME_GROSS_CONVERSIONS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAffiliate(affId, begin, end, item, helper::addGrossConversions));
        stringRedisTemplate.boundZSetOps(REAL_TIME_SCRUB_CONVERSIONS).rangeWithScores(0, -1)
                .forEach(item -> setValueForAffiliate(affId, begin, end, item, helper::addScrubConversions));

        return helper.data();
    }

    @Override
    public List<RealTimeData> mergeRealTimeData(AllRedisRealTimeData allRedisRealTimeData, List<RealTimeGroupByField> fields,
            RealTimeDataSearchParam param, Instant begin, Instant end) {
        final RealTimeDataHelper helper = new RealTimeDataHelper();
        if (allRedisRealTimeData != null) {
            allRedisRealTimeData.getRealTimeClicks()
                    .forEach(item -> setValue(fields, param, begin, end, item, helper::addClicks));
            allRedisRealTimeData.getRealTimeUclicks()
                    .forEach(item -> setValue(fields, param, begin, end, item, helper::addUClicks));
            allRedisRealTimeData.getRealTimeGrossConversions()
                    .forEach(item -> setValue(fields, param, begin, end, item, helper::addGrossConversions));
            allRedisRealTimeData.getRealTimeScrubConversions()
                    .forEach(item -> setValue(fields, param, begin, end, item, helper::addScrubConversions));
        }
        return helper.data();
    }

    private void setValue(List<RealTimeGroupByField> fields,
            RealTimeDataSearchParam param, Instant begin, Instant end,
            ZSetOperations.TypedTuple<String> item,
            BiConsumer<RealTimeKey, Integer> addMethod) {
        /*
         * 聚合数据不同的纬度,取决于生成的RealTimeKey字段的值,
         * 只设置了offerId,则根据offer进行聚合,设置了offerId和affId,则根据这两个字段进行聚合
         */
        RealTimeKey realTimeKey = RealTimeKey.fromKey(item.getValue(), fields);
        String realTimeKeyJson = JsonMapper.nonEmptyMapper().toJson(realTimeKey);
        logger.info("realTimeKey:{}", realTimeKeyJson);
        if (realTimeKey != null && realTimeKey.isMatch(fields, param, begin, end)) {
            int score = item.getScore().intValue();
            addMethod.accept(realTimeKey, score);
        }
    }

    private void setValueForAdvertiser(int advId, Instant begin, Instant end,
            ZSetOperations.TypedTuple<String> item,
            BiConsumer<RealTimeKey, Integer> addMethod) {
        RealTimeKey realTimeKey = RealTimeKey.fromKey(item.getValue());
        if (realTimeKey != null && realTimeKey.isMatchForAdvertiser(advId, begin, end)) {
            addMethod.accept(realTimeKey, item.getScore().intValue());
        }
    }

    private void setValueForAffiliate(int affId, Instant begin, Instant end,
            ZSetOperations.TypedTuple<String> item,
            BiConsumer<RealTimeKey, Integer> addMethod) {
        RealTimeKey realTimeKey = RealTimeKey.fromKey(item.getValue());
        if (realTimeKey != null && realTimeKey.isMatchForAffiliate(affId, begin, end)) {
            addMethod.accept(realTimeKey, item.getScore().intValue());
        }
    }

    @Override
    public AllRedisRealTimeData allRedisRealTimeData() {
        Set<ZSetOperations.TypedTuple<String>> realTimeClicks =
                stringRedisTemplate.boundZSetOps(REAL_TIME_CLICKS).rangeWithScores(0, -1);
        Set<ZSetOperations.TypedTuple<String>> realTimeUclicks =
                stringRedisTemplate.boundZSetOps(REAL_TIME_UCLICKS).rangeWithScores(0, -1);
        Set<ZSetOperations.TypedTuple<String>> realTimeGrossConversions =
                stringRedisTemplate.boundZSetOps(REAL_TIME_GROSS_CONVERSIONS).rangeWithScores(0, -1);
        Set<ZSetOperations.TypedTuple<String>> realTimeScrubConversions =
                stringRedisTemplate.boundZSetOps(REAL_TIME_SCRUB_CONVERSIONS).rangeWithScores(0, -1);
        logger.info("realTimeClicks:{},realTimeUclicks:{},realTimeGrossConversions:{},realTimeScrubConversions:{}",
                realTimeClicks.size(), realTimeUclicks.size(), realTimeGrossConversions.size(), realTimeScrubConversions.size());
        AllRedisRealTimeData allRedisRealTimeData = new AllRedisRealTimeData();
        allRedisRealTimeData.setRealTimeClicks(realTimeClicks);
        allRedisRealTimeData.setRealTimeUclicks(realTimeUclicks);
        allRedisRealTimeData.setRealTimeGrossConversions(realTimeGrossConversions);
        allRedisRealTimeData.setRealTimeScrubConversions(realTimeScrubConversions);
        return allRedisRealTimeData;
    }
}
